package org.example.fboot;

import lombok.extern.slf4j.Slf4j;
import org.example.fboot.proto.MessageRequestOuterClass;
import org.example.fboot.proto.MessageResponseOuterClass;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Component
public abstract class MessageHandler<Message extends MessageRequestOuterClass.MessageRequest> {
    private final Map<String, Map<String, WebSocketSession>> sessionMap = new ConcurrentHashMap<>();

    /**
     * 上线
     *
     * @param wsId
     * @param webSocketSession
     */
    public void session(String wsId, WebSocketSession webSocketSession) {
        Map<String, WebSocketSession> room = sessionMap.getOrDefault(path(), new ConcurrentHashMap<>());
        sessionMap.put(path(), room);
        log.info("在线人数:{} 上线：{} 当前上线:{}", room.size(), wsId, webSocketSession.getId());
        //挤下线
        WebSocketSession session = room.get(wsId);
        room.put(wsId, webSocketSession);
        if (session != null) {
            session.close().subscribe();
        }
        log.info("上线：在线人数：{}", room.size());
        log.info("上线：房间数：{}", sessionMap.size());
    }

    /**
     * 下线
     *
     * @param wsId
     * @param session
     */
    public void out(String wsId, WebSocketSession session) {
        Map<String, WebSocketSession> room = sessionMap.getOrDefault(path(), new ConcurrentHashMap<>());
        if (!room.isEmpty()) {
            WebSocketSession socketSession = room.get(wsId);
            if (socketSession == session) {
                room.remove(wsId);
            }
        }
        log.error("离线：在线人数：{}", room.size());
    }

    public void sender(Message message) {
        switch (message.getMessageType()) {
            case ALL -> sendAll(sessionMap.get(path()), message);
            case PERSONS -> sendTo(sessionMap.get(path()), message);
        }
    }

    /**
     * 给房间里某人通知
     *
     * @param message
     */
    private void sendTo(Map<String, WebSocketSession> room, Message message) {
        send(message.getRecipientList(), room, message);
    }

    /**
     * 发送给房间里所有人
     *
     * @param message
     */
    private void sendAll(Map<String, WebSocketSession> room, Message message) {
        send(room.keySet(), room, message);
    }

    private void send(Iterable<String> recipient, Map<String, WebSocketSession> room, Message message) {
        Flux.fromIterable(recipient)
                .filter(m -> !m.equalsIgnoreCase(message.getSender()))
                .doOnNext(m->{
                    log.info("发送数据:发送方：{},接收方：{}",message.getSender(),m);
                })
                .flatMap(m -> Mono.fromCallable(() -> buildMessage(m, room, message)))
                .doOnNext((e -> e.k.send(Mono.just(e.v)).subscribe()))
                .doOnError(e -> log.error("发送信息异常：", e))
                .subscribe();
    }

    private Entity<WebSocketSession, WebSocketMessage> buildMessage(String recipient, Map<String, WebSocketSession> room, Message message) {
        MessageResponseOuterClass.MessageResponse build = MessageResponseOuterClass.MessageResponse.newBuilder()
                .setRecipient(recipient)
                .setBodyTypeValue(message.getBodyTypeValue())
                .setMessageTypeValue(message.getMessageTypeValue())
                .setSender(message.getSender())
                .setData(message.getData())
                .build();
        WebSocketSession session = room.get(recipient);
        WebSocketMessage binaryMessage = session.binaryMessage(b -> b.wrap(build.toByteArray()));
        return new Entity<>(session, binaryMessage);
    }

    public abstract String path();


    private record Entity<K, V>(K k, V v) {
    }
}
